package cn.doitedu.rtdw.dashboard;

import cn.doitedu.rtdw.beans.OmsOrder;
import cn.doitedu.rtdw.beans.OmsOrderCdc;
import com.alibaba.fastjson.JSON;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import io.debezium.data.Json;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

public class Job06_最近一小时的各类订单数_API版 {

    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);


        // 创建source读数据: mysql-cdc的source
        MySqlSource<String> cdcSource = MySqlSource.<String>builder()
                .databaseList("realtimedw")
                .tableList("realtimedw.oms_order")
                .hostname("doitedu")
                .port(3306)
                .username("root")
                .password("root")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();

        DataStreamSource<String> streamSource = env.fromSource(cdcSource, WatermarkStrategy.noWatermarks(), "cdc-source");

        // 处理
        // 数据解析
        SingleOutputStreamOperator<OmsOrderCdc> cdcBeanStream = streamSource.map(new MapFunction<String, OmsOrderCdc>() {
            @Override
            public OmsOrderCdc map(String json) throws Exception {
                return JSON.parseObject(json, OmsOrderCdc.class);
            }
        });
        SingleOutputStreamOperator<OmsOrderCdc> watermarked = cdcBeanStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<OmsOrderCdc>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(new SerializableTimestampAssigner<OmsOrderCdc>() {
                            @Override
                            public long extractTimestamp(OmsOrderCdc cdcBean, long recordTimestamp) {

                                OmsOrder omsOrder;
                                if (cdcBean.getOp().equals("d")) {
                                    omsOrder = cdcBean.getBefore();
                                } else {
                                    omsOrder = cdcBean.getAfter();
                                }
                                return omsOrder.getModify_time();
                            }
                        }));

        // 求最近一小时内创建的各状态订单数
        // 开滑动窗口进行计算
        watermarked
                .windowAll(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(1)))
                .process(new ProcessAllWindowFunction<OmsOrderCdc, String, TimeWindow>() {


                    MapState<Integer, Integer> statusOrderCntState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        statusOrderCntState = getRuntimeContext().getMapState(new MapStateDescriptor<Integer, Integer>("status_order_cnt", Integer.class, Integer.class));
                    }

                    @Override
                    public void process(ProcessAllWindowFunction<OmsOrderCdc, String, TimeWindow>.Context context, Iterable<OmsOrderCdc> elements, Collector<String> out) throws Exception {
                        // 取出本窗口的起始、结束时间
                        long windowStart = context.window().getStart()-8*60*60*1000;
                        long windowEnd = context.window().getEnd()-8*60*60*1000;

                        // 遍历本窗口中的每一条cdc日志
                        for (OmsOrderCdc cdcBean : elements) {

                            OmsOrder before = cdcBean.getBefore();
                            OmsOrder after = cdcBean.getAfter();
                            // 取出订单的创建时间
                            long create_time = after == null ? before.getCreate_time() : after.getCreate_time();
                            create_time = create_time - 8*60*60*1000;

                            // 判断该条订单操作日志，是否是本时间窗口内创建的订单
                            if (create_time < windowStart || create_time > windowEnd) return;

                            /* *
                             * 统计状态1 的订单数
                             */
                            if ((before == null || before.getStatus() != 1) && after.getStatus() == 1) {
                                // 从非1，变成1，对状态1的订单数+
                                Integer oldCount = statusOrderCntState.get(1);
                                int oldCnt = oldCount == null ? 0 : oldCount;

                                statusOrderCntState.put(1, oldCnt + 1);
                            }
                            if ((before != null && before.getStatus() == 1) && (after == null || after.getStatus() != 1)) {
                                // 从1，变成非1，对状态1的订单数-
                                Integer oldCount = statusOrderCntState.get(1);
                                int oldCnt = oldCount == null ? 0 : oldCount;

                                statusOrderCntState.put(1, oldCnt - 1);
                            }

                            /* *
                             * 统计状态2 的订单数
                             */

                        }


                        HashMap<Object, Object> tmpMap = new HashMap<>();
                        tmpMap.put("window_start", DateFormatUtils.format(windowStart,"yyyy-MM-dd HH:mm:ss.SSS"));
                        tmpMap.put("window_end",DateFormatUtils.format(windowEnd,"yyyy-MM-dd HH:mm:ss.SSS"));

                        for (Map.Entry<Integer, Integer> entry : statusOrderCntState.entries()) {
                            tmpMap.put(entry.getKey(), entry.getValue());
                        }

                        String resultJson = JSON.toJSONString(tmpMap);
                        out.collect(resultJson);

                        // 要清除状态，否则这个状态会在后续的窗口中不断延续
                        statusOrderCntState.clear();


                    }
                })
                .print();



        env.execute();
    }

}
